package com.wisecoach.gatewayplus.event;

import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * {@code @author:} wisecoach
 * {@code @date:} 2023/6/15 上午12:47
 * {@code @version:} 1.0.0
 */
public class DefaultFilteredBlockEventPublisher implements FilteredBlockEventPublisher {

    private final FilteredBlockEventMulticaster multicaster;
    private final FilteredBlockSourceRegistry registry;
    private final Map<String, Set<FilteredBlockEventListener>> channelToListeners = new ConcurrentHashMap<>();

    public DefaultFilteredBlockEventPublisher(FilteredBlockEventMulticaster multicaster, FilteredBlockSourceRegistry registry) {
        this.multicaster = multicaster;
        this.registry = registry;
    }

    @Override
    public void publish(FilteredBlockEvent event) {
        multicaster.multicastEvent(event);
    }

    @Override
    public void start() {
        Map<String, Long> channelsToStartNum = getChannelsToStartNum();
        channelsToStartNum.forEach((channel, startNumber) -> {
            FilteredBlockSource source = registry.getSource(channel);
            if (source == null) {
                throw new EventException("不存在要监听的区块源，通道为：" + channel);
            }
            source.setFilteredBlockEventPublisher(this);
            source.start(startNumber, null);
        });
    }

    @Override
    public void addListener(String channel, FilteredBlockEventListener listener) {
        FilteredBlockSource source = registry.getSource(channel);
        if (source == null) {
            throw new EventException("不存在要监听的区块源，通道为：" + channel);
        }
        Set<FilteredBlockEventListener> listeners = channelToListeners.computeIfAbsent(channel, k -> new LinkedHashSet<>());
        listeners.add(listener);
        switch (listener.getListenerResumeType()) {
            case FIRST:
                multicaster.addListener(channel, listener, 0);
                break;
            case CONTINUE:
                multicaster.addListener(channel, listener);
            case LATEST:
                multicaster.addListener(channel, listener, source.getLatest());
            default:
                multicaster.addListener(channel, listener);
        }
    }

    @Override
    public void removeListener(String channel, FilteredBlockEventListener listener) {
        Set<FilteredBlockEventListener> listeners = channelToListeners.get(channel);
        if (listeners != null) {
            listeners.remove(listener);
        }
        multicaster.removeListener(channel, listener);
    }

    @Override
    public void clearListener() {
        channelToListeners.clear();
        multicaster.clearListener();
    }

    private Map<String, Long> getChannelsToStartNum() {
        HashMap<String, Long> channelsToStartNum = new HashMap<>();
        channelToListeners.forEach((channel, listeners) -> {
            // 找出开始区块最小的监听器，加入channel和最小区块号
            listeners.stream().min(
                    (listener1, listener2) -> Math.toIntExact(listener1.getNext() - listener2.getNext())
            ).ifPresent((listener) -> channelsToStartNum.put(channel, listener.getNext()));
        });
        return channelsToStartNum;
    }

}
